package org.ytor.common.util.concurrent;


import lombok.extern.slf4j.Slf4j;
import org.ytor.common.enums.DateUnit;
import org.ytor.common.exception.BaseException;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * created by yangtong on 2024/6/9 23:46:48
 * 多线程工具类，封装线程池与调度器，支持虚拟线程、高效并发与定时任务调度
 */
@Slf4j
public class T {

    /**
     * 物理线程池
     */
    private final static ThreadPoolExecutor physicalPool;

    /**
     * 虚拟线程池
     */
    private final static ExecutorService virtualPool;

    /**
     * 调度线程，用来定时给线程池提交任务
     */
    private final static ScheduledExecutorService scheduler;

    /**
     * CPU核心数量
     */
    private final static int CPU_CORES = Runtime.getRuntime().availableProcessors();

    /**
     * 任务处理失败的处理逻辑
     */
    private static final List<ExceptionHandler> exceptionHandlers = new ArrayList<>();

    private static final AtomicInteger THREAD_COUNTER = new AtomicInteger(0);

    static {
        //物理线程适合CPU密集型
        physicalPool = new ThreadPoolExecutor(
                CPU_CORES,
                (int) (CPU_CORES * 1.5),
                60L,  // 建议增加空闲时间
                TimeUnit.SECONDS,  // 改为秒
                new LinkedBlockingQueue<>(1024),
                r -> new Thread(r, "physical-worker-" + THREAD_COUNTER.incrementAndGet()),
                new ThreadPoolExecutor.CallerRunsPolicy()  // 建议改为调用者运行策略
        );

        //虚拟线程适合IO阻塞场景
        virtualPool = Executors.newVirtualThreadPerTaskExecutor();

        //调度线程只需要单线程即可
        scheduler = Executors.newSingleThreadScheduledExecutor(
                r -> new Thread(r, "scheduler-" + THREAD_COUNTER.incrementAndGet())
        );
    }

    /**
     * 注册任务处理失败的处理逻辑
     */
    public static void registerExceptionHandler(ExceptionHandler handler) {
        exceptionHandlers.add(handler);
    }

    // ==================== 睡眠工具方法 ====================

    /**
     * 线程睡眠
     */
    public static void sleep(long duration) {
        sleep(duration, DateUnit.MS);
    }

    /**
     * 线程睡眠
     */
    public static void sleep(long duration, DateUnit timeUnit) {
        try {
            Thread.sleep(duration * timeUnit.getMillis());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 不响应中断的sleep
     */
    public static void sleepUnInterrupted(long time) {
        boolean interrupted = false;
        try {
            long remaining = time;
            long start = System.currentTimeMillis();
            while (remaining > 0) {
                try {
                    Thread.sleep(remaining);
                } catch (InterruptedException e) {
                    interrupted = true;
                }
                remaining = time - (System.currentTimeMillis() - start);
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // ==================== 物理线程任务执行方法 ====================

    /**
     * 使用物理线程执行没有返回值任务
     */
    public static void exec(Runnable task) {
        physicalPool.execute(wrap(task));
    }

    /**
     * 使用物理线程提交有返回值的任务
     */
    public static <T> Future<T> exec(Callable<T> task) {
        return physicalPool.submit(wrap(task));
    }

    /**
     * 使用物理线程执行没有返回值多个任务，只有任务全部完成，才会继续执行后续代码
     */
    public static void execBatch(Runnable... tasks) {
        execBatchInternal(physicalPool, tasks);
    }

    /**
     * 使用物理线程周期性调度任务
     */
    public static ScheduledFuture<?> execScheduled(Runnable task, Long period, DateUnit dateUnit) {
        return execScheduledInternal(physicalPool, task, period, dateUnit);
    }

    /**
     * 使用物理线程，限次数调度
     */
    public static ScheduledFuture<?> execScheduledLimited(Runnable task, Long period, DateUnit unit, Integer maxTimes) {
        return execScheduledLimitedInternal(physicalPool, task, period, unit, maxTimes);
    }

    /**
     * 用物理线程延时执行一次任务
     */
    public static ScheduledFuture<?> execOnce(Runnable task, long delay, DateUnit dateUnit) {
        return execOnceInternal(physicalPool, task, delay, dateUnit);
    }

    // ==================== 虚拟线程任务执行方法 ====================

    /**
     * 使用虚拟线程执行没有返回值任务
     */
    public static void execVirtual(Runnable task) {
        virtualPool.execute(wrap(task));
    }

    /**
     * 使用虚拟线程提交有返回值的任务（修复bug）
     */
    public static <T> Future<T> execVirtual(Callable<T> task) {
        return virtualPool.submit(wrap(task));  // 修复：使用virtualPool而不是physicalPool
    }

    /**
     * 使用虚拟线程执行没有返回值多个任务
     */
    public static void execBatchVirtual(Runnable... tasks) {
        execBatchInternal(virtualPool, tasks);
    }

    /**
     * 使用虚拟线程周期性调度任务
     */
    public static ScheduledFuture<?> execScheduledVirtual(Runnable task, Long period, DateUnit dateUnit) {
        return execScheduledInternal(virtualPool, task, period, dateUnit);
    }

    /**
     * 使用虚拟线程，限次数调度
     */
    public static ScheduledFuture<?> execScheduledLimitedVirtual(Runnable task, Long period, DateUnit unit, Integer maxTimes) {
        return execScheduledLimitedInternal(virtualPool, task, period, unit, maxTimes);
    }

    /**
     * 用虚拟线程延时执行一次任务
     */
    public static ScheduledFuture<?> execOnceVirtual(Runnable task, long delay, DateUnit dateUnit) {
        return execOnceInternal(virtualPool, task, delay, dateUnit);
    }


    /**
     * 批量执行任务的内部实现
     */
    private static void execBatchInternal(ExecutorService executor, Runnable... tasks) {
        try {
            CountDownLatch latch = new CountDownLatch(tasks.length);

            for (Runnable task : tasks) {
                executor.execute(() -> {
                    try {
                        wrap(task).run();
                    } finally {
                        latch.countDown();
                    }
                });
            }

            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BaseException(e);
        }
    }

    /**
     * 周期性调度任务的内部实现
     */
    private static ScheduledFuture<?> execScheduledInternal(ExecutorService executor, Runnable task, Long period, DateUnit dateUnit) {
        long millis = dateUnit.getMillis();
        return scheduler.scheduleWithFixedDelay(() -> {
            executor.execute(wrap(task));
        }, 0, period * millis, TimeUnit.MILLISECONDS);
    }

    /**
     * 限次数调度的内部实现
     */
    private static ScheduledFuture<?> execScheduledLimitedInternal(ExecutorService executor, Runnable task, Long period, DateUnit unit, Integer maxTimes) {
        AtomicInteger counter = new AtomicInteger(0);
        final ScheduledFuture<?>[] futureRef = new ScheduledFuture<?>[1];

        long millis = period * unit.getMillis();

        futureRef[0] = scheduler.scheduleWithFixedDelay(() -> {
            int count = counter.incrementAndGet();
            if (count > maxTimes) {
                futureRef[0].cancel(false);
                return;
            }
            executor.execute(wrap(task));
        }, 0, millis, TimeUnit.MILLISECONDS);

        return futureRef[0];
    }

    /**
     * 延时执行一次任务的内部实现
     */
    private static ScheduledFuture<?> execOnceInternal(ExecutorService executor, Runnable task, long delay, DateUnit dateUnit) {
        long delayMillis = delay * dateUnit.getMillis();
        return scheduler.schedule(() -> executor.execute(wrap(task)), delayMillis, TimeUnit.MILLISECONDS);
    }


    /**
     * 监控物理线程池状态
     */
    public static void logThreadPoolStatus() {
        logPhysicalPoolStatus();
        logVirtualPoolStatus();
    }

    /**
     * 监控物理线程池状态
     */
    public static void logPhysicalPoolStatus() {
        int corePoolSize = physicalPool.getCorePoolSize();
        int poolSize = physicalPool.getPoolSize();
        int activeCount = physicalPool.getActiveCount();
        int largestPoolSize = physicalPool.getLargestPoolSize();
        long completedTaskCount = physicalPool.getCompletedTaskCount();
        int queueSize = physicalPool.getQueue().size();

        log.info("[物理线程池监控] 核心线程数: {}, 当前线程数: {}, 活跃线程数: {}, 峰值线程数: {}, " +
                        "已完成任务数: {}, 队列中的任务数: {}",
                corePoolSize, poolSize, activeCount, largestPoolSize, completedTaskCount, queueSize);
    }

    /**
     * 监控虚拟线程池状态
     */
    public static void logVirtualPoolStatus() {
        log.info("[虚拟线程池监控] 虚拟线程池正在运行，详细监控需要通过JVM工具查看");
    }

    /**
     * 带重试的任务执行
     */
    public static <T> T execWithRetry(Callable<T> task, int maxRetries, long delayMs) {
        Exception lastException = null;

        for (int i = 0; i <= maxRetries; i++) {
            try {
                return task.call();
            } catch (Exception e) {
                lastException = e;
                if (i < maxRetries) {
                    log.warn("任务执行失败，第{}次重试，错误：{}", i + 1, e.getMessage());
                    sleep(delayMs);
                } else {
                    log.error("任务执行失败，已达到最大重试次数：{}", maxRetries);
                }
            }
        }

        throw new BaseException("任务执行失败，重试" + maxRetries + "次后仍然失败", lastException);
    }

    /**
     * 带指数退避的重试机制
     */
    public static <T> T execWithExponentialBackoff(Callable<T> task, int maxRetries, long initialDelayMs, double multiplier) {
        Exception lastException = null;
        long currentDelay = initialDelayMs;

        for (int i = 0; i <= maxRetries; i++) {
            try {
                return task.call();
            } catch (Exception e) {
                lastException = e;
                if (i < maxRetries) {
                    log.warn("任务执行失败，第{}次重试，延迟{}ms，错误：{}", i + 1, currentDelay, e.getMessage());
                    sleep(currentDelay);
                    currentDelay = (long) (currentDelay * multiplier);
                } else {
                    log.error("任务执行失败，已达到最大重试次数：{}", maxRetries);
                }
            }
        }

        throw new BaseException("任务执行失败，重试" + maxRetries + "次后仍然失败", lastException);
    }

    /**
     * 等待条件满足
     */
    public static void waitFor(Supplier<Boolean> condition, long timeoutMs, long checkIntervalMs) {
        long startTime = System.currentTimeMillis();

        while (!condition.get()) {
            if (System.currentTimeMillis() - startTime > timeoutMs) {
                throw new BaseException("等待超时");
            }
            sleep(checkIntervalMs);
        }
    }

    /**
     * 异步等待条件满足
     */
    public static CompletableFuture<Void> waitForAsync(Supplier<Boolean> condition, long checkIntervalMs) {
        return CompletableFuture.runAsync(() -> {
            while (!condition.get()) {
                sleep(checkIntervalMs);
            }
        }, virtualPool);
    }

    /**
     * 优雅关闭所有线程池
     */
    public static void shutdown() {
        log.info("开始关闭线程池...");

        // 关闭调度器
        scheduler.shutdown();

        // 关闭物理线程池
        physicalPool.shutdown();

        // 关闭虚拟线程池
        if (virtualPool != null) {
            try {
                ((AutoCloseable) virtualPool).close();
            } catch (Exception e) {
                log.warn("关闭虚拟线程池时发生异常", e);
            }
        }

        try {
            // 等待任务完成
            if (!physicalPool.awaitTermination(30, TimeUnit.SECONDS)) {
                physicalPool.shutdownNow();
            }
            if (!scheduler.awaitTermination(30, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            physicalPool.shutdownNow();
            scheduler.shutdownNow();
        }

        log.info("线程池关闭完成");
    }

    private static Runnable wrap(Runnable task) {
        String submitter = Thread.currentThread().getName();
        return () -> {
            long start = System.currentTimeMillis();
            try {
                task.run();
            } catch (Throwable e) {
                log.error("线程任务执行异常：{}", e.getMessage(), e);
                exceptionHandlers.forEach(handler -> handler.handle(e, submitter));
            } finally {
                long cost = System.currentTimeMillis() - start;
                if (log.isDebugEnabled()) {
                    log.debug("任务执行 [{}] 完成，由 [{}] 提交，耗时 [{}ms]", Thread.currentThread().getName(), submitter, cost);
                }
            }
        };
    }

    public static <V> Callable<V> wrap(Callable<V> task) {
        String submitter = Thread.currentThread().getName();
        return () -> {
            long start = System.currentTimeMillis();
            String resultStr = null;
            try {
                V result = task.call();
                resultStr = String.valueOf(result);
                if (resultStr.length() > 500) {
                    resultStr = resultStr.substring(0, 500) + "...(truncated)";
                }
                return result;
            } catch (Throwable e) {
                log.error("线程任务执行异常：{}", e.getMessage(), e);
                exceptionHandlers.forEach(handler -> handler.handle(e, submitter));
                throw e;
            } finally {
                long cost = System.currentTimeMillis() - start;
                if (log.isDebugEnabled()) {
                    log.debug("任务执行 [{}] 完成，由 [{}] 提交，耗时 [{}ms]，返回结果：[{}]",
                            Thread.currentThread().getName(), submitter, cost, resultStr);
                }
            }
        };
    }
}
